package com.example.mq.mqserver.core;

import com.example.mq.common.Consumer;
import com.example.mq.common.ConsumerEnv;
import com.example.mq.common.MqException;
import com.example.mq.mqserver.VirtualHost;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.BlockingDeque;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingDeque;


@Slf4j
/*
 * 通过这个类来实现消费消息
 * 每当有消息到队列的时候,就将队列名(令牌)添加到令牌队列中,线程池再根据这个队列进行处理
 */
public class ConsumerManager {
    //方便操作数据
    private VirtualHost parent;
    //使用线程池,用来处理回调任务
    private ExecutorService workPool = Executors.newFixedThreadPool(4);
    //每当队列中来了消息,就触发这个令牌队列,通知可以去消费消息
    private BlockingDeque<String> tokenQueue = new LinkedBlockingDeque<>();
    //扫描线程,扫描令牌队列中是否有新令牌
    private Thread scanThread = null;

    public ConsumerManager(VirtualHost p){
        this.parent = p;
        //创建扫描线程
        scanThread = new Thread(() ->{
            while(true){
                try {
                    //获取令牌
                    String tokenQueueName = tokenQueue.take();
                    //根据令牌找到指定队列
                    MSGQueue queue = parent.getMemoryDataCenter().getQueue(tokenQueueName);
                    if(queue == null){
                        throw new MqException("[ConsumerManager] 令牌队列不存在 queueName:" + tokenQueueName);
                    }
                    //从队列中消费消息
                    synchronized (queue){
                        consumeMessage(queue);
                    }
                } catch (InterruptedException | MqException e) {
                    e.printStackTrace();
                }
            }
        });

        //设置为后台线程
        scanThread.setDaemon(true);
        scanThread.start();
    }


    //通知消费消息
    public void notifyConsumer(String queueName) throws InterruptedException {
        tokenQueue.put(queueName);
    }

    //添加订阅者
    public void addConsumer(String consumerTag, String queueName, boolean autoAck, Consumer consumer){
        MSGQueue queue = parent.getMemoryDataCenter().getQueue(queueName);
        try{
            if(queue == null){
                throw new MqException("[ConsumerManager] 队列不存在 queueName:" + queueName);
            }
            synchronized (queue){
                //添加消费者
                queue.addConsumerEnv(new ConsumerEnv(consumerTag, queueName, autoAck, consumer));
                //如果有消息就进行消费
                int n = parent.getMemoryDataCenter().getMessageSize(queueName);
                for(int i = 0; i < n; i++){
                    //消费
                    consumeMessage(queue);
                }
            }
        }catch (Exception e){
            log.error("[ConsumerManager] 添加订阅者失败");
            e.printStackTrace();
        }
    }

    //消费消息
    private void consumeMessage(MSGQueue queue){
        //1.通过轮询方式,寻找消费者
        ConsumerEnv luckDog = queue.chooseConsumer();
        if(luckDog == null){
            return;
        }
        //2.获取队列中的消息
        Message message = parent.getMemoryDataCenter().pollMessage(queue.getName());
        if(message == null){
            return;
        }
        //3.把消息带入到回调方法,交给线程池处理
        workPool.submit(() -> {
            try{
                //1.把消息添加到待确认队列中
                parent.getMemoryDataCenter().addMessageWaitAck(queue.getName(), message);
                //2.执行回调操作,消费消息
                luckDog.getConsumer().handleDelivery(luckDog.getConsumerTag(), message.getBasicProperties(), message.getBody());
                //3.如果当前是自动应答.则将消息删除
                //  如果是手动应答,则先不处理,让消费者后续调用 basicAck 处理
                if(luckDog.isAutoAck()){
                    //删除消息从两个方面进行,内存和硬盘
                    //硬盘
                    if(message.isDurable()){
                        parent.getDiskDataCenter().deleteMessage(queue, message);
                    }
                    //内存
                    parent.getMemoryDataCenter().deleteMessageWaitAck(queue.getName(), message.getMessageId());
                    parent.getMemoryDataCenter().removeMessage(message.getMessageId());
                    log.info("[ConsumerManager] 消息被成功消费 queueName:" + queue.getName());
                }
            }catch (Exception e){
                log.info("[ConsumerManager] 消息消费失败 queueName:" + queue.getName());
                e.printStackTrace();
            }
        });
    }
}
